-- SET sql-client.execution.result-mode=TABLEAU;
-- 执行命令： sql-client.sh -f flink_kafka2mysql.q

CREATE TABLE tbl_host_zzdetail (
  content STRING,
  content2 STRING
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://hllxd101:53149/db58_zp_bi_dev',
  'username' = 'zpbi_wr',
  'password' = '39086120578ddd85',
  'table-name' = 'tbl_host_zzdetail'
);

CREATE TABLE hdp_lbg_supin_ods_tsar_detail2 (
  content STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'host_zzdetail2',
  'properties.bootstrap.servers' = 'node101:9092',
--  'scan.startup.mode' = 'group-offsets',
  'scan.startup.mode' = 'earliest-offset',
  'sink.parallelism' = '1',
  'format' = 'raw'
);

INSERT INTO tbl_host_zzdetail
SELECT
  content, PROCTIME() as content2
FROM hdp_lbg_supin_ods_tsar_detail2

INSERT INTO tbl_host_zzdetail
SELECT
  content, cast(PROCTIME() as string) as content2
FROM hdp_lbg_supin_ods_tsar_detail2



INSERT INTO tbl_hism_host_zzdetail2
SELECT
  content, cast(PROCTIME() as string) as content2
FROM hdp_lbg_supin_ods_tsar_detail2
